package com.mjf.orderpay_detect

import java.net.URL

import com.mjf.dim.{OrderEvent, ReceiptEvent}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object OrderPayTxMatchWithJoin {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val orderResource: URL = getClass.getResource("/OrderLog.csv")
    val orderStream: DataStream[String] = env.readTextFile(orderResource.getPath)

    // 订单数据
    val orderEventStream: KeyedStream[OrderEvent, String] = orderStream
      .map {
        line =>
          val dataArr: Array[String] = line.split(",")
          OrderEvent(dataArr(0).toLong, dataArr(1), dataArr(2), dataArr(3).toLong)
      }
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.seconds(3)) {
        override def extractTimestamp(element: OrderEvent): Long = element.eventTime * 1000L
      })
      .filter(_.eventType != "") // 过滤出pay事件
      .keyBy(_.txId)

    val receiptResource: URL = getClass.getResource("/ReceiptLog.csv")
    val receiptStream: DataStream[String] = env.readTextFile(receiptResource.getPath)

    // 实时到账数据
    val ReceiptEventStream: KeyedStream[ReceiptEvent, String] = receiptStream
      .map {
        line =>
          val dataArr: Array[String] = line.split(",")
          ReceiptEvent(dataArr(0), dataArr(1), dataArr(2).toLong)
      }
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ReceiptEvent](Time.seconds(3)) {
        override def extractTimestamp(element: ReceiptEvent): Long = element.timestamp * 1000L
      })
      .keyBy(_.txId)

    // 使用join连接两条流
    val resultStream: DataStream[(OrderEvent, ReceiptEvent)] = orderEventStream
      .intervalJoin(ReceiptEventStream)
      .between(Time.seconds(-3), Time.seconds(5))
      .process(new OrderPayTxDetectWithJoin())

    resultStream.print()

    env.execute("OrderPayTxMatchWithJoin")

  }
}

class OrderPayTxDetectWithJoin extends ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)] {
  override def processElement(left: OrderEvent, right: ReceiptEvent, ctx: ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    out.collect(left, right)
  }
}